package com.zhanggm.message.message.tast;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.bw.springboot.stater.email.bean.MailBean;
import com.bw.springboot.stater.email.service.MailService;
import com.zhanggm.message.message.entity.KafkaLog;
import com.zhanggm.message.message.service.IKafkaLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.UUID;

/**
 * 定时任务处理失败消息
 */
@Component
@EnableScheduling
@Slf4j
public class MsgTask {
    @Autowired
    private IKafkaLogService iKafkaLogService;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Autowired
    private MailService mailService;
     //检查失败消息，30秒检查一次
//    @Scheduled(cron = "30 * * * * ?")
    @Scheduled(fixedDelay = 5000)
    public void msgCheck() {
        log.info("msgCheck");
        //查询失败消息
        QueryWrapper queryWrapper = new QueryWrapper();
        queryWrapper.eq("msg_status", 2);
        queryWrapper.le("msg_count", 1);
        List<KafkaLog> list = iKafkaLogService.list(queryWrapper);
        //补偿，重新发送邮件
        list.forEach(kafkaLog -> {
            String msgContent = kafkaLog.getMsgContent();
            MailBean mailBean = JSON.parseObject(msgContent, MailBean.class);
            mailService.sendSimpleMail(mailBean);
//            kafkaTemplate.send("hetongEmailTopic", JSON.toJSONString(mailBean));
            //修改消息状态
            kafkaLog.setMsgStatus(1);
            kafkaLog.setMsgCount(kafkaLog.getMsgCount()+1);
            iKafkaLogService.saveOrUpdate(kafkaLog);
        });
    }
}
